/*
 * BackendNode.java
 *
 * Version:
 *      $Id: BackendNode.java 27 2009-04-17 02:30:39Z PErickson87 $
 *
 * Copyright (c) 2009 Peter O. Erickson
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without major
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, and/or distribute copies of, but
 * under no circumstances sublicense and/or sell, the Software,
 * and to permit persons to whom the Software is furnished to do
 * so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 */

package edu.rit.poe.cluster;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.CyclicBarrier;

/**
 * A JCluster backend node application.  This backend node service will listen
 * and accept connections on the well-known port of the Communicator, receive
 * class information, and run a single node of an application.
 *
 * @author  Peter O. Erickson
 *
 * @version $Id: BackendNode.java 27 2009-04-17 02:30:39Z PErickson87 $
 */
public class BackendNode {
    
    /** The default port for listening for connections from the job server. */
    public static final int JOB_SERVER_PORT = 4241;
    
    /** The default port to connect to a node Communicator. */
    public static final int COMM_PORT = 4242;
    
    /** The default port for listening to shutdown commands. */
    public static final int SHUTDOWN_PORT = 4244;
    
    private static boolean running;
    
    private static Job job;
    
    private static int jobCount;
    
    /**
     * The main method that runs the backend node server.
     * 
     * @param   args        the command line arguments
     */
    public static void main( String[] args ) {
        if ( args.length < 2 ) {
            
        }
        
        jobCount = 0;
        
        // setup the NIO selector
        Selector selector = null;
        try {
            selector = Selector.open();
        } catch ( IOException e ) {
            System.err.println( "Error setting up NIO selector." );
            System.exit( -1 );
        }
        
        // setup the ports and sockets to listen on
        ServerSocketChannel server = null;
        try {
            // listen on 4241 for Job Server
            server = ServerSocketChannel.open();
            server.socket().bind( new InetSocketAddress( JOB_SERVER_PORT ) );
            server.configureBlocking( false );
            server.register( selector, SelectionKey.OP_ACCEPT,
                    JOB_SERVER_PORT );
            
            // listen on 4242 for shutdown
            server = ServerSocketChannel.open();
            server.socket().bind( new InetSocketAddress( SHUTDOWN_PORT ) );
            server.configureBlocking( false );
            server.register( selector, SelectionKey.OP_ACCEPT, SHUTDOWN_PORT );
            
        } catch ( IOException e ) {
            System.err.println( "Error registering server socket channels with "
                    + " NIO selector." );
            System.exit( -1 );
        }
        
        // start a thread to listen (traditional java.net) for Comm connections
        running = true;
        new Thread() {
            @Override
            public void run() {
                ServerSocket server = null;
                try {
                    server = new ServerSocket( COMM_PORT );
                } catch ( IOException e ) {
                    System.err.println( "Error starting server." );
                    System.exit( -1 );
                }
                Socket socket = null;
                while ( running ) {
                    try {
                        System.out.println( "Waiting for job..." );
                        socket = server.accept();
                        server.close();
                    } catch ( IOException e ) {
                        System.err.println(
                                "Error accepting job from Comm listener." );
                    }
                    if ( job == null ) {
                        job = new Job( socket );
                        job.start();
                        try {
                            job.join();
                        } catch ( InterruptedException e ) {
                            // problem
                        }
                        job.kill();
                        job = null;
                        System.out.println( "Finished job " + jobCount++ );
                    } else {
                        //TODO get angry at requestor, or queue
                    }
                    
                    // restart the Comm listener
                    try {
                        server = new ServerSocket( COMM_PORT );
                    } catch ( IOException e ) {
                        // this is unrecoverable
                        System.err.println( "Error restarting Comm listener." );
                        System.exit( -1 );
                    }
                }
            }
        }.start();
        
        // run until we're shutdown
        while ( running ) {
            try {
                selector.select();
            } catch ( IOException e ) {
                System.err.println( "Error selecting." );
                System.exit( -1 );
            }
            
            Set<SelectionKey> keys = selector.keys();
            Iterator<SelectionKey> iter = keys.iterator();
            
            while ( iter.hasNext() ) {
                SelectionKey key = iter.next();
                Object att = key.attachment();
                SelectableChannel channel = key.channel();
                
                if ( key.isAcceptable() ) {
                    // we are accepting on one of the server socket channels
                    Integer i = ( Integer )att;
                    
                    if ( i == JOB_SERVER_PORT ) {
                        
                        // accept connection, register for read listening
                        
                    } else if ( i == SHUTDOWN_PORT ) {
                        if ( job != null ) {
                            job.interrupt();
                            job.kill();
                            job = null;
                        }
                        running = false;
                        break;
                    }
                }
            }
        }
        
        // force quit (non-daemon threads could still exist, exit forcibly)
        System.exit( 0 );
    }
    
    private static class Job extends Thread {
        
        private Socket socket;
        
        Job( Socket socket ) {
            this.socket = socket;
        }
        
        @Override
        public void run() {
            String name = null;
            String[] rootargs = null;
            try {
                DataInputStream is = new DataInputStream(
                        socket.getInputStream() );
                ObjectInputStream ois = new ObjectInputStream( is );
                
                // receive: [class name] (UTF String)
                name = is.readUTF();
                
                // receive: [size] (int)
                Comm.size = is.readInt();
                
                // reiceve: [rank] (int)
                Comm.rank = is.readInt();
                
                // receive: [String[] args object]
                rootargs = ( String[] )ois.readObject();
                
                // receive: [String[] nodehosts List]
                Comm.nodehosts = ( String[] )ois.readObject();
                
                // initialize barrier, nodes and root
                Comm.barrier = new CyclicBarrier( Comm.size );
                Comm.nodes = new Comm.Node[ Comm.size ];
                Comm.nodes[ 0 ] = new Comm.Node( ( short )0,
                        socket );
            } catch ( IOException e ) {
                System.err.println(
                        "Error receiving job information." );
            } catch ( ClassNotFoundException e ) {
                // this would only happen if we can't find String[]
                assert false;
            }

            // setup the network classloader
            InetAddress addr = socket.getInetAddress();
            String host = addr.getHostAddress();
            NetworkClassLoader ncl = null;
            try {
                ncl = new NetworkClassLoader( host );
            } catch ( UnknownHostException e ) {
                System.err.println( "Invalid hostname of network "
                        + "class loader server." );
            }
            Comm.ncl = ncl;

            // load the class to run
            Class clazz = null;
            Method method = null;
            try {
                clazz = ncl.loadClass( name );
                method = clazz.getMethod( "main", rootargs.getClass() );
            } catch ( Exception e ) {
                System.err.println( "Error loading the main class "
                        + "from the root node." );
            }

            // run the node process
            // Comm.init() will establish interconnectivity on each node
            try {
                method.invoke( null, new Object[]{ rootargs } );
            } catch ( Exception e ) {
                System.err.println( "Uncaught exception while "
                        + "executing application." );
                e.printStackTrace();
            }
        }
        
        void kill() {
            // send end signal to the root node, shutdown nodes, and restart
            // comm port server
            try {
                DataOutputStream out = new DataOutputStream(
                        socket.getOutputStream() );
                out.writeByte( Comm.Node.Command.NODE_END );
                out.close();
                socket.close();  

                // shutdown all node connections
                Comm.shutdown();
            } catch ( IOException e ) {
                System.err.println( "Error sending node end." );
            }
        }
        
    } // Job
    
} // BackendNode
